package dm.ass2.item;

import java.io.IOException;
import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Set;

import org.apache.commons.lang.ArrayUtils;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import dm.ass2.ClusteringDriver;
import dm.ass2.io.CodebookItemsWritable;
import dm.ass2.io.CodebookUsersWritable;
import dm.ass2.item.io.ItemMapOutputValueWritable;
import dm.ass2.user.UserDriver;
import dm.ass2.user.io.CBElement;
import dm.ass2.user.io.UserMapOutputValueWritable;

/**
 * @author yaron
 */
public class ItemReducer extends Reducer<IntWritable, ItemMapOutputValueWritable, Text, CodebookItemsWritable> {
	List<CodebookItemsWritable> cbItems;
	int numOfItemClusters = 2;
	int numOfUserClusters = 2;

	@Override
	protected void cleanup(Context context) throws IOException, InterruptedException {
		super.cleanup(context);
		Path cbItemssPath = new Path(ClusteringDriver.OUTPUT_DIR + ClusteringDriver.CB_ITEMS_FILENAME);
		FileSystem fs;

		fs = FileSystem.get(context.getConfiguration());

		if (fs.exists(cbItemssPath)) {
			fs.delete(cbItemssPath, true);
		}

		FSDataOutputStream out = fs.create(cbItemssPath);
		for (CodebookItemsWritable cbItem : cbItems) {
			cbItem.write(out);
		}
		out.flush();
		out.close();
		
	}

	@Override
	protected void reduce(IntWritable itemClusterId, Iterable<ItemMapOutputValueWritable> values, Context context) throws IOException, InterruptedException {
		Set<Integer> itemIds = new HashSet<Integer>();
		CBElement[] ratings = new CBElement[numOfItemClusters];
		
		for (ItemMapOutputValueWritable itemMapOutput : values) {
			for (int currItemId : itemMapOutput.getUsersList()) {
				itemIds.add(currItemId);
			}

			CBElement[] itemCbElements = itemMapOutput.getCbElements();
			for (int i = 0; i < itemCbElements.length; i++) {
				CBElement currElement;
				if (ratings[i] == null) {
					currElement = new CBElement(0, 0f);
					ratings[i] = currElement;
				} else {
					currElement = ratings[i];
				}
				currElement.increment(itemCbElements[i].getRatingsCounter());
				currElement.addToSum(itemCbElements[i].getRatingsSum());
			}
		}
		
		float[] clusterItemsRatings = new float[ratings.length];
		
		for (int i = 0; i < ratings.length; i++) {
			clusterItemsRatings[i] = ratings[i].avg();
		}
		
		int[] itemIdsArray = ArrayUtils.toPrimitive(itemIds.toArray(new Integer[itemIds.size()]));
		
		CodebookItemsWritable cbItem = new CodebookItemsWritable(itemClusterId.get(), itemIdsArray, clusterItemsRatings);
		cbItems.add(cbItem);
	}

	@Override
	protected void setup(Context context) throws IOException, InterruptedException {
		super.setup(context);
		this.numOfItemClusters = Integer.parseInt(context.getConfiguration().get(UserDriver.NUM_OF_ITEM_CLUSTERS));
		this.numOfUserClusters = Integer.parseInt(context.getConfiguration().get(UserDriver.NUM_OF_USER_CLUSTERS));
		
		cbItems = new LinkedList<CodebookItemsWritable>();
	}
	
	
}
